package database.hset;

import database.inter.IDB;
import database.singleDB.SingleDB;
import datastruct.dict.impl.JoConcurrentMap;
import datastruct.dict.inter.IDict;
import datastruct.dict.inter.IDictConsumer;
import jodis.conn.impl.*;
import jodis.conn.inter.IReply;
import rdb.RDBConstant;
import tcp.Server;

import java.math.BigDecimal;
import java.security.PublicKey;
import java.util.Arrays;

public class JHashSet {
    public static IDict getAsDict(IDB idb, String key) {
        if (idb instanceof SingleDB) {
            SingleDB db =  (SingleDB) idb;
            Object obj = db.getEntity(key);
            return (IDict) obj;
        }
        return null;
    }

    public static IDict getOrInitDict(IDB idb, String key) {
        IDict dict = getAsDict(idb, key);
        if (dict == null && idb instanceof SingleDB) {
            dict = JoConcurrentMap.makeConcurrentMap();
            SingleDB db = (SingleDB) idb;
            db.putEntity(key, dict);
        }
        return dict;
    }

//    HSET key field value [field value ...]
    public static IReply execHSet(IDB idb, String[] cmdLines) {
        String key = cmdLines[1];
        IDict dict = getOrInitDict(idb, key);
        // field value 不是成对的
        if ((cmdLines.length-2) % 2 != 0) {
            return StandardErrReply.makeErrReply("wrong number of argument");
        }
        int size = (cmdLines.length-2) / 2;
        int j = 0, ret = 0;
        for (int i=0;i<size;i++,j+=2) {
            String field = cmdLines[2+j];
            String value = cmdLines[2+j+1];
            dict.put(field, value);
            ret++;
        }

        SingleDB db = (SingleDB) idb;
        db.putType(key, RDBConstant.TYPE_HASH);

        if (Server.getServerConfig().appendOnly) {
            db.iAddAof.addAof(db.index, cmdLines);
        }

        return IntReply.makeIntReply(ret);
    }
    //    HSET key field value [field value ...]
    public static String[][] undoHSet(IDB idb, String[] cmdLines) {
        String key = cmdLines[1];
        IDict dict = getAsDict(idb, key);
        // 在执行 hset 之前，这个 key 不存在，那么回滚时，也要删除这个 key
        if (dict == null) {
            return new String[][]{
                {"Del", key}
            };
        }
        int size = (cmdLines.length-2)/2;
        int j = 2;
        String [][] undoCmdLines = new String[size][];
        for (int i=0;i<size;i++,j+=2) {
            // 这 hset 执行前，这个 field 是不存在的，那么回滚的时候，这个 field 应该还原成不存在
            if (dict.get(cmdLines[j]) == null) {
                // hdel key field [field]...
                undoCmdLines[i] = new String[]{
                        "HDel", key, cmdLines[j]
                };
            } else {
                // 如果这个 field 之前就存在，那么就还原成原来的值
                Object oldObj = dict.get(cmdLines[j]);
                undoCmdLines[i] = new String[]{
                        "HSet", key, cmdLines[j], oldObj.toString()
                };
            }
        }
        return undoCmdLines;
    }

    // HGet key field
    public static IReply execHGet(IDB idb, String[] cmdLines) {
        String key = cmdLines[1];
        IDict dict = getAsDict(idb, key);
        if (dict == null) {
            return EmptyMultiBulkReply.makeEmptyMultiBulkReply();
        }

        String res = (String) dict.get(cmdLines[2]);
        if (res == null) {
            return NullReply.makeNullReply();
        }

        return MultiBulkReply.makeMultiBulkReply(new String[]{res});
    }

//    HDEL key field [field ...]
    public static IReply execHDel(IDB idb, String[] cmdLines) {
        String key = cmdLines[1];
        IDict dict = getAsDict(idb, key);
        if (dict == null) {
            return IntReply.makeIntReply(0);
        }

        int ret = 0;
        for (int i=2;i<cmdLines.length;i++) {
            ret += dict.remove(cmdLines[i]);
        }

        SingleDB db = (SingleDB) idb;

        if (dict.len() == 0) {
            db.removeEntity(key);
        }

        if (Server.getServerConfig().appendOnly) {
            db.iAddAof.addAof(db.index, cmdLines);
        }

        return IntReply.makeIntReply(ret);
    }

//    HDEL key field [field ...]
    public static String[][] undoHDel(IDB idb, String[] cmdLines) {
        String key = cmdLines[1];
        IDict dict = getAsDict(idb, key);
        // 在 HDel 执行前，这个 key 对应的 dict 就不存在，所以回滚的时候，也应该删除这个 key
        if (dict == null) {
             return new String[][]{
                     {"Del", key}
             };
        }
//        hset key field value
        String[][] undoCmdLines = new String[cmdLines.length-2][];
        int j = 0;
        for (int i=2;i<cmdLines.length;i++) {
            // 在 HDel 执行前，这个 field 存在，那么回滚时，也应该包含这个field 和 field 对应的 val
            if (dict.contains(cmdLines[i])) {
                Object oldObj = dict.get(cmdLines[i]);
                undoCmdLines[j] = new String[]{
                        "HSet", key, cmdLines[i], oldObj.toString()
                };
                j++;
            }
        }
        return undoCmdLines;
    }

    // SCARD key
    public static IReply execSCard(IDB idb, String[] cmdLines) {
        IDict dict = getAsDict(idb, cmdLines[1]);
        return dict == null ? IntReply.makeIntReply(0) : IntReply.makeIntReply(dict.len());
    }

//    SISMEMBER key member
    public static IReply execSIsMember(IDB idb, String[] cmdLines) {
        IDict dict = getAsDict(idb, cmdLines[1]);
        int res = dict == null ? 0 : dict.contains(cmdLines[2]) ? 1 : 0;
        return IntReply.makeIntReply(res);
    }

//    SRANDMEMBER key [count]
    public static IReply execSRandMember(IDB idb, String[] cmdLines) {
        IDict dict = getAsDict(idb, cmdLines[1]);
        // key 不存在，且没指定 count，那么返回 nil，如果指定 count，那么返回 empty array
        if (dict == null) {
            return cmdLines.length == 3 ? EmptyMultiBulkReply.makeEmptyMultiBulkReply() : NullReply.makeNullReply();
        }

        // 没指定 count
        if (cmdLines.length == 2) {
            return MultiBulkReply.makeMultiBulkReply(dict.randomKeys(1));
        }

        // 指定了 count
         try {
             int count = Integer.parseInt(cmdLines[2]);
             System.out.printf("count: %d\n", count);
             return count > 0 ? MultiBulkReply.makeMultiBulkReply(dict.randomDistinctKeys(count)) : MultiBulkReply.makeMultiBulkReply(dict.randomKeys(-count));
         } catch (NumberFormatException e) {
             e.printStackTrace();
             return StandardErrReply.makeErrReply(e.getMessage());
         }
    }

    // SPOP key [count]
    public static IReply execSPop(IDB idb, String[] cmdLines) {
        // key 不存在，没指定 count，返回 nil，指定了 count，返回 empty array
        IDict dict = getAsDict(idb, cmdLines[1]);
        if (dict == null) {
            return cmdLines.length != 3 ? NullReply.makeNullReply() : EmptyMultiBulkReply.makeEmptyMultiBulkReply();
        }

        int count = 1;
        // 指定了 count
        if (cmdLines.length == 3) {
            try {
                count = Integer.parseInt(cmdLines[2]);
                if (count < 0) {
                    return StandardErrReply.makeErrReply("value is out of range, must be positive");
                }
            } catch (NumberFormatException e) {
                return StandardErrReply.makeErrReply("illegal count");
            }
        }

        count = Math.abs(Math.min(count, dict.len()));
        String[] items = new String[count];
        String[] keys = dict.randomDistinctKeys(count);
        int i = 0;
        for (String k : keys) {
            items[i] = k;
            dict.remove(k);
            i++;
        }

        return MultiBulkReply.makeMultiBulkReply(items);
    }


    // SPOP key [count]
    public static String[][] undoSPop(IDB idb, String[] cmdLines) {
        IDict dict = getAsDict(idb, cmdLines[1]);
        if (dict == null) {
            return new String[][]{
                    {"Del", cmdLines[1]}
            };
        }

        // 因为是随机删除某个元素，所以回滚的时候，直接添加所有元素
        if (dict.len() == 0) {
            return new String[][]{};
        }

        String[] keys = dict.keys();
        String[] items = new String[2+keys.length];
        items[0] = "SAdd";
        items[1] = cmdLines[1];

        int i = 2;
        for (String k : keys) {
            items[i++] = k;
        }

        return new String[][]{
                items
        };
    }

//    SDIFF key [key ...]
    // 找第一个 key 对应的 set 与其他 key 对应的 set 合集 的 diff
    public static IReply execSDiff(IDB idb, String[] cmdLines) {
        if (cmdLines.length < 2 ) {
            return StandardErrReply.makeErrReply("illegal number of arguments");
        }

        IDict firstSet = getAsDict(idb, cmdLines[1]);
        if (firstSet == null || firstSet.len() == 0) {
            return EmptyMultiBulkReply.makeEmptyMultiBulkReply();
        }

        if (cmdLines.length == 2) {
            return MultiBulkReply.makeMultiBulkReply(firstSet.keys());
        }

        IDict[] dts = new IDict[cmdLines.length-2];
        for (int i=2, j=0;i<cmdLines.length;i++,j++) {
            dts[j] = getAsDict(idb, cmdLines[i]);
        }

        String[] res = (String[]) firstSet.diff(dts);
        return MultiBulkReply.makeMultiBulkReply(res);
    }

    public static String[] prepareSDiff(String[] cmdLines) {
        if (cmdLines.length < 2) {
            return new String[]{};
        }
        return Arrays.copyOfRange(cmdLines, 1, cmdLines.length);
    }

    public static String[] prepareSUnion(String[] cmdLines) {
        return prepareSDiff(cmdLines);
    }

//    SUNION key [key ...]
    public static IReply execSUnion(IDB idb, String[] cmdLines) {
        if (cmdLines.length < 2) {
            return StandardErrReply.makeErrReply("illegal number of arguments");
        }

        IDict firstSet = getAsDict(idb, cmdLines[1]);
        if (firstSet == null || firstSet.len() == 0) {
            return EmptyMultiBulkReply.makeEmptyMultiBulkReply();
        }

        if (cmdLines.length == 2) {
            return MultiBulkReply.makeMultiBulkReply(firstSet.keys());
        }


        IDict[] dts = new IDict[cmdLines.length-2];
        for (int i=2, j=0;i<cmdLines.length;i++,j++) {
            dts[j] = getAsDict(idb, cmdLines[i]);
        }

        String[] res = (String[]) firstSet.union(dts);
        return MultiBulkReply.makeMultiBulkReply(res);
    }

//    SAdd key member [member]
    public static IReply execSAdd(IDB idb, String[] cmdLines) {
        String key = cmdLines[1];
        int ret = 0;
        IDict dict = getOrInitDict(idb, key);
        for (int i=2;i<cmdLines.length;i++) {
            ret += dict.put(cmdLines[i], true);
        }
        SingleDB db = (SingleDB) idb;
        db.putType(key, RDBConstant.TYPE_SET);
        if (Server.getServerConfig().appendOnly) {
            db.iAddAof.addAof(db.index, cmdLines);
        }
        return IntReply.makeIntReply(ret);
    }

    public static String[][] undoSAdd(IDB idb, String[] cmdLines) {
        String key = cmdLines[1];
        IDict dict = getAsDict(idb, key);
        // 在执行 sadd 之前，这个 key 不存在，那么回滚时，也要删除这个 key
        if (dict == null) {
            return new String[][]{
                    {"Del", key}
            };
        }
        String[][] undoCmdLines = new String[cmdLines.length-2][];
        int j = 0;
        for (int i=2;i<cmdLines.length;i++,j++) {
            // 这 sadd 执行前，这个 member 是不存在的，那么回滚的时候，这个 member 应该还原成不存在
            if (dict.get(cmdLines[i]) == null) {
                undoCmdLines[j] = new String[]{
                        "SRem", key, cmdLines[i]
                };
            } else {
                // 如果这个 member 之前就存在，那么就还原成原来的值
                undoCmdLines[j] = new String[]{
                        "SAdd", key, cmdLines[i]
                };
            }
        }
        return undoCmdLines;
    }

//    SREM key member [member ...]
    public static IReply execSRem(IDB idb, String[] cmdLines) {
        String key = cmdLines[1];
        IDict dict = getAsDict(idb, key);
//        这个 key 不存在
        if (dict == null) {
            return EmptyMultiBulkReply.makeEmptyMultiBulkReply();
        }
        int ret = 0;
        for (int i=2;i<cmdLines.length;i++) {
            if (dict.contains(cmdLines[i])) {
                ret++;
                dict.remove(cmdLines[i]);
            }
        }
        SingleDB db = (SingleDB) idb;
        if (dict.len() == 0) {
            db.removeEntity(key);
        }

        if (Server.getServerConfig().appendOnly) {
            db.iAddAof.addAof(db.index, cmdLines);
        }
        return IntReply.makeIntReply(ret);
    }

//    SREM key member [member ...]
    // SREM myset2 "one"
    public static String[][] undoSRem(IDB idb, String[] cmdLines) {
        String key = cmdLines[1];
        IDict dict = getAsDict(idb, key);
        // 在执行 srem 之前，这个 key 不存在，那么回滚时，也要删除这个 key
        if (dict == null) {
            return new String[][]{
                    {"Del", key}
            };
        }

        String[][] undoCmdLines = new String[cmdLines.length-2][];
        for (int i=2,j=0;i<cmdLines.length;i++,j++) {
            // 这 srem 执行前，这个 member 是不存在的，那么回滚的时候，这个 member 应该还原成不存在
            if (!dict.contains(cmdLines[i])) {
                undoCmdLines[j] = new String[]{
                        "SRem", key, cmdLines[i]
                };
            } else {
                // 如果这个 member 之前就存在，那么就还原成原来的值
                undoCmdLines[j] = new String[]{
                        "SAdd", key, cmdLines[i]
                };
            }
        }
        return undoCmdLines;
    }

    public static IReply execSMembers(IDB idb, String[] cmdLines) {
        String key = cmdLines[1];
        IDict dict = getAsDict(idb, key);
        if (dict == null) {
            return EmptyMultiBulkReply.makeEmptyMultiBulkReply();
        }

        String[] res = new String[dict.len()];
        final int[] i = {0};
        dict.forEach(new IDictConsumer() {
            @Override
            public boolean consumer(String key, Object val) {
                Boolean v = (Boolean) val;
                if (v) {
                    res[i[0]] = key;
                    i[0]++;
                }
                return true;
            }
        });
        return MultiBulkReply.makeMultiBulkReply(res);
    }

//    HSETNX key field value
    public static IReply execHSetNX(IDB idb, String[] cmdLines) {
        IDict dict = getOrInitDict(idb, cmdLines[1]);
        String field = cmdLines[2];

        return IntReply.makeIntReply(dict.putIfAbsent(field, cmdLines[3]));
    }

    public static String[][] undoHSetNX(IDB idb, String[] cmdLines) {
        IDict dict = getAsDict(idb, cmdLines[1]);
        if (dict == null || dict.len() == 0) {
            return new String[][]{
                    {"Del", cmdLines[1]}
            };
        }
        // 如果 field 已经存在，那么什么也不用做
        if (dict.contains(cmdLines[2])) {
            return new String[][]{};
        }

        return new String[][]{
                {"HDel", cmdLines[1], cmdLines[2]}
        };
    }

//    HEXISTS key field
    public static IReply execHExists(IDB idb, String[] cmdLines) {
        String key = cmdLines[1];
        IDict dict = getAsDict(idb, key);
        int res = dict == null || !dict.contains(cmdLines[2]) ? 0 : 1;
        return IntReply.makeIntReply(res);
    }

//    HLEN key
    public static IReply execHLen(IDB idb, String[] cmdLines) {
        String key = cmdLines[1];
        IDict dict = getAsDict(idb, key);
        int res = dict == null ? 0 : dict.len();
        return IntReply.makeIntReply(res);
    }

//    HSTRLEN key field
    public static IReply execHStrlen(IDB idb, String[] cmdLines) {
        if (cmdLines.length != 3) {
            return StandardErrReply.makeErrReply("illegal number of arguments");
        }
        String key = cmdLines[1];
        IDict dict = getAsDict(idb, key);
        int res = dict == null || !dict.contains(cmdLines[2]) ? 0 : ((String) dict.get(cmdLines[2])).length();
        return IntReply.makeIntReply(res);
    }

//    HMSET key field value [field value ...]
    // As of Redis version 4.0.0, this command is regarded as deprecated.
    public static IReply execHMSet(IDB idb, String[] cmdLines) {
        if (cmdLines.length % 2 != 0 || cmdLines.length == 2) {
            return StandardErrReply.makeErrReply("illegal arguments for HMSet");
        }

        String key = cmdLines[1];
        IDict dict = getOrInitDict(idb, key);
        for (int i=2;i<cmdLines.length-1;i+=2) {
            String field = cmdLines[i];
            String value = cmdLines[i+1];
            dict.put(field, value);
        }

        return OkReply.makeOKReply();
    }
    //    HMSET key field value [field value ...]
    public static String[][] undoHMSet(IDB idb, String[] cmdLines) {
        if (cmdLines.length % 2 != 0 || cmdLines.length == 2) {
            return new String[][]{};
        }

        IDict dict = getAsDict(idb, cmdLines[1]);
        if (dict == null) {
            return new String[][]{
                    {"Del", cmdLines[1]}
            };
        }

        String[] items = new String[2+(cmdLines.length-2)/2];
        items[0] = "HDel";
        items[1] = cmdLines[1];
        for (int i=2, j=2;i<cmdLines.length-1;i+=2,j++) {
            String field = cmdLines[i];
            items[j] = field;
        }

        return new String[][]{items};
    }

//    HMGET key field [field ...]
    public static IReply execHMGet(IDB idb, String[] cmdLines) {
        if (cmdLines.length < 3) {
            return StandardErrReply.makeErrReply("illegal arguments for HMGet");
        }

        IDict dict = getAsDict(idb, cmdLines[1]);
        String[] res = new String[cmdLines.length-2];
        if (dict == null) {
//            Arrays.fill(res, NullReply.NULL_STRING);
            return MultiBulkReply.makeMultiBulkReply(res);
        }

        for (int i=2,j=0;i<cmdLines.length;i++,j++) {
            res[j] = dict.contains(cmdLines[i]) ? (String) dict.get(cmdLines[i]) : null;
        }

        return MultiBulkReply.makeMultiBulkReply(res);
    }

    // HKeys key
    public static IReply execHKeys(IDB idb, String[] cmdLines) {
        if (cmdLines.length < 2) {
            return StandardErrReply.makeErrReply("illegal arguments for HKeys");
        }
        IDict dict = getAsDict(idb, cmdLines[1]);
        if (dict == null) {
            return EmptyMultiBulkReply.makeEmptyMultiBulkReply();
        }

        return MultiBulkReply.makeMultiBulkReply(dict.keys());
    }

//    HVALS key
    public static IReply execHVals(IDB idb, String[] cmdLines) {
        if (cmdLines.length < 2) {
            return StandardErrReply.makeErrReply("illegal arguments for HVals");
        }
        IDict dict = getAsDict(idb, cmdLines[1]);
        if (dict == null) {
            return EmptyMultiBulkReply.makeEmptyMultiBulkReply();
        }

        String[] res = new String[dict.len()];
        final int[] i = {0};
        dict.forEach(new IDictConsumer() {
            @Override
            public boolean consumer(String key, Object val) {
                res[i[0]++] = (String) val;
                return true;
            }
        });

        return MultiBulkReply.makeMultiBulkReply(res);
    }

//    HGETALL key
    public static IReply execHGetAll(IDB idb, String[] cmdLines) {
        if (cmdLines.length < 2) {
            return StandardErrReply.makeErrReply("illegal arguments for HGetAll");
        }
        IDict dict = getAsDict(idb, cmdLines[1]);
        if (dict == null) {
            return EmptyMultiBulkReply.makeEmptyMultiBulkReply();
        }

        String[] res = new String[dict.len()*2];
        final int[] i = {0};

        dict.forEach(new IDictConsumer() {
            @Override
            public boolean consumer(String key, Object val) {
                res[i[0]] = key;
                res[i[0]+1] = (String) val;
                i[0] += 2;
                return true;
            }
        });

        return MultiBulkReply.makeMultiBulkReply(res);
    }

//    HINCRBY key field increment
    public static IReply execHIncrBy(IDB idb, String[] cmdLines) {
        IDict dict = getOrInitDict(idb, cmdLines[1]);
        if (!dict.contains(cmdLines[2])) {
            dict.put(cmdLines[2], cmdLines[3]);
            try {
                int res = Integer.parseInt(cmdLines[3]);
                return IntReply.makeIntReply(res);
            } catch (NumberFormatException e) {
                return StandardErrReply.makeErrReply(e.getMessage());
            }
        }

        String oldObj = (String) dict.get(cmdLines[2]);
        try {
            int val = Integer.parseInt(oldObj);
            int incr = Integer.parseInt(cmdLines[3]);
            val += incr;
            dict.put(cmdLines[2], Integer.toString(val));
            return MultiBulkReply.makeMultiBulkReply(new String[]{Integer.toString(val)});
        } catch (NumberFormatException e) {
            return StandardErrReply.makeErrReply(e.getMessage());
        }
    }

    //    HINCRBY key field increment
    public static String[][] undoHIncrBy(IDB idb, String[] cmdLines) {
        IDict dict = getAsDict(idb, cmdLines[1]);
        if (dict == null) {
            return new String[][]{
                    {"Del", cmdLines[1]}
            };
        }

        if (!dict.contains(cmdLines[2])) {
            return new String[][]{
                    {"HDel", cmdLines[1], cmdLines[2]}
            };
        }

        String oldObj = (String) dict.get(cmdLines[2]);
        return new String[][]{
                {"HSet", cmdLines[1], cmdLines[2], oldObj}
        };
    }

//    HINCRBYFLOAT key field increment
    public static IReply execHIncrByFloat(IDB idb, String[] cmdLines) {
        IDict dict = getOrInitDict(idb, cmdLines[1]);
        if (!dict.contains(cmdLines[2])) {
            dict.put(cmdLines[2], cmdLines[3]);
            try {
                double res = Double.parseDouble(cmdLines[3]);
                return MultiBulkReply.makeMultiBulkReply(new String[]{Double.toString(res)});
            } catch (NumberFormatException e) {
                return StandardErrReply.makeErrReply(e.getMessage());
            }
        }

        String oldObj = (String) dict.get(cmdLines[2]);

        try {
            BigDecimal val = new BigDecimal(oldObj);
            BigDecimal incr = new BigDecimal(cmdLines[3]);
            val = val.add(incr);
//            if (incr.compareTo(BigDecimal.ZERO) < 0) {
//                val = val.subtract(incr);
//            } else {
//                val = val.add(incr);
//            }
            dict.put(cmdLines[2], val.toString());
            return MultiBulkReply.makeMultiBulkReply(new String[]{val.toString()});
        } catch (NumberFormatException e) {
            return StandardErrReply.makeErrReply(e.getMessage());
        }

    }

    public static String[][] undoHIncrByFloat(IDB idb, String[] cmdLines) {
        return undoHIncrBy(idb, cmdLines);
    }

//    HRANDFIELD key [count [WITHVALUES]]
    public static IReply execHRandField(IDB idb, String[] cmdLines) {
        int count = 1;
        boolean withValues = cmdLines.length == 4 && cmdLines[3].equalsIgnoreCase("withvalues");

        if (cmdLines.length > 2) {
            try {
                count = Integer.parseInt(cmdLines[2]);
            } catch (NumberFormatException e) {
                return StandardErrReply.makeErrReply("illegal count");
            }
        }

        IDict dict = getAsDict(idb, cmdLines[1]);
        if (dict == null) {
            return EmptyMultiBulkReply.makeEmptyMultiBulkReply();
        }

        String[] keys = count > 0 ? dict.randomDistinctKeys(count) : dict.randomKeys(-count);
        String[] res = withValues ? new String[keys.length*2] : null;

        if (withValues) {
            for (int i=0,j=0;i<res.length-1;i+=2,j++) {
                res[i] = keys[j];
                Object val = dict.get(keys[j]);
                res[i+1] = val == null ? "null" : (String)val;
            }
        }
        return withValues ? MultiBulkReply.makeMultiBulkReply(res) : MultiBulkReply.makeMultiBulkReply(keys);
    }
}
